package org.artifact.core.plugin.activemq;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.function.Consumer;

import javax.jms.BytesMessage;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;

import org.apache.activemq.spring.ActiveMQConnectionFactory;
import org.artifact.core.lang.IPlugin;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.listener.SessionAwareMessageListener;

import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;

public class ActiveMQPlugin implements IPlugin {

	static final Log log = LogFactory.get(ActiveMQPlugin.class);
	// 链接对象
	// private ActiveMQConnection connection;

	// 事务管理
	// private ActiveMQSession session;

	CachingConnectionFactory cachingConnectionFactory;

	// 队列生产者
	private Map<String, JmsTemplate> queueJmsTemplate = new HashMap<>();
	// 队列监听器
	private Map<String, SessionAwareMessageListener> queueListeners = new HashMap<>();

	// 广播生产者
	private Map<String, JmsTemplate> topicJmsTemplate = new HashMap<>();
	// 广播监听器
	private Map<String, SessionAwareMessageListener> topicListeners = new HashMap<>();

	private List<DefaultMessageListenerContainer> listenerContainer = new ArrayList<>();

	// private MessageConverter messageConverter;

	private String userName;
	private String password;
	private String brokenUrl;
	private boolean init = false;

	private volatile boolean isClose = false;

	public void init(String userName, String password, String brokenUrl) {
		this.userName = userName;
		this.password = password;
		this.brokenUrl = brokenUrl;

		// ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(userName,
		// password, brokenUrl);
		// poolFactory = new PooledConnectionFactory(factory);
		// // 池中借出的对象的最大数目
		// poolFactory.setMaxConnections(100);
		// poolFactory.setMaximumActiveSessionPerConnection(50);
		// //后台对象清理时，休眠时间超过了3000毫秒的对象为过期
		// poolFactory.setTimeBetweenExpirationCheckMillis(3000);
		init = true;
	}

	public void addListener(boolean topic, String queueName, SessionAwareMessageListener Listener) {
		if (topic) {
			topicListeners.put(queueName, Listener);
		} else {
			queueListeners.put(queueName, Listener);
		}
	}

	@Override
	public boolean start() {
		if (!init) {
			throw new IllegalArgumentException("Plugin No initialization");
		}

		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
		connectionFactory.setBrokerURL(brokenUrl);
		connectionFactory.setUserName(userName);
		connectionFactory.setPassword(password);
		connectionFactory.setUseAsyncSend(true);

		cachingConnectionFactory = new CachingConnectionFactory(connectionFactory);
		cachingConnectionFactory.setSessionCacheSize(8);
		// 转换器
		// messageConverter = new MessageConverter() {
		// @Override
		// public Message toMessage(Object arg0, Session arg1) throws JMSException,
		// MessageConversionException {
		// return (Message) arg0;
		// }
		// @Override
		// public Object fromMessage(Message arg0) throws JMSException,
		// MessageConversionException {
		// return arg0;
		// }
		// };
		// 创建监听器
		for (Entry<String, SessionAwareMessageListener> e : queueListeners.entrySet()) {
			DefaultMessageListenerContainer listener = new DefaultMessageListenerContainer();
			listener.setConnectionFactory(cachingConnectionFactory);
			listener.setDestinationName(e.getKey());
			listener.setMessageListener(e.getValue());
			listener.setPubSubDomain(false);
			// listener.setMessageConverter(messageConverter);
			listener.initialize();
			listener.start();
			listenerContainer.add(listener);
		}

		// 创建监听器
		for (Entry<String, SessionAwareMessageListener> e : topicListeners.entrySet()) {
			DefaultMessageListenerContainer listener = new DefaultMessageListenerContainer();
			listener.setConnectionFactory(cachingConnectionFactory);
			listener.setDestinationName(e.getKey());
			listener.setMessageListener(e.getValue());
			listener.setPubSubDomain(true);
			// listener.setMessageConverter(messageConverter);
			listener.initialize();
			listener.start();
			listenerContainer.add(listener);
		}

		return true;
	}

	public boolean isClose() {
		return isClose;
	}

	@Override
	public boolean stop() {
		if (!isClose) {
			for (DefaultMessageListenerContainer defaultMessageListenerContainer : listenerContainer) {
				defaultMessageListenerContainer.stop();
			}
		}
		return true;
	}

	// 异步发送消息
	public void asynSendMessage(boolean topic, String queueName, Message msg) throws JMSException {
		if (isClose) {
			throw new JMSException("MQ is closed!");
		}
//		GameExecutors.getThreadPoolExecutor(GameExecutors.ExecutorType.activeMQ).execute(() -> {
			// 取得一个生产者
			JmsTemplate jmsTemplate = null;
			if (topic) {
				jmsTemplate = topicJmsTemplate.get(queueName);
				if (jmsTemplate == null) {
					jmsTemplate = createTopicJmsTemplate(queueName);
				}
			} else {
				jmsTemplate = queueJmsTemplate.get(queueName);
				if (jmsTemplate == null) {
					jmsTemplate = createQueueJmsTemplate(queueName);
				}
			}
			jmsTemplate.convertAndSend(msg);
//		});
	}

	public void syncSendMessage(String queueName, Message msg, MessageListener replyListen) throws JMSException {
		if (isClose) {
			throw new JMSException("MQ is closed!");
		}
		JmsTemplate jmsTemplate = queueJmsTemplate.get(queueName);
		if (jmsTemplate == null) {
			jmsTemplate = createQueueJmsTemplate(queueName);
		}

		Message message = jmsTemplate.sendAndReceive((Session session) -> {
			// BytesMessage byteMsg = session.createBytesMessage();
			// byteMsg.writeBytes(SerializationUtil.serialize(msg));
			return msg;
		});
		if (message == null) {
			throw new JMSException("receive timeout!");
		}
		replyListen.onMessage(message);
	}

	public Message syncSendMessage(String queueName, Message msg) throws JMSException {
		if (isClose) {
			throw new JMSException("MQ is closed!");
		}
		JmsTemplate jmsTemplate = queueJmsTemplate.get(queueName);
		if (jmsTemplate == null) {
			jmsTemplate = createQueueJmsTemplate(queueName);
		}
		Message message = jmsTemplate.sendAndReceive((Session session) -> {
			// BytesMessage byteMsg = session.createBytesMessage();
			// byteMsg.writeBytes(SerializationUtil.serialize(msg));
			return msg;
		});
		if (message == null) {
			throw new JMSException("receive timeout!");
		}
		return message;
	}
	
	/**
	 * 返回消息
	 * @param session
	 * @param req
	 * @param consumer
	 */
	public void result(Session session, Message req,Consumer<Message> consumer) {
		try {
			MessageProducer producer = session.createProducer(null);
			// 创建返回消息
			BytesMessage resp = session.createBytesMessage();
			resp.setJMSCorrelationID(req.getJMSCorrelationID());
			// 消息封装
			consumer.accept(resp);
			producer.send(req.getJMSReplyTo(),resp);
		} catch (Exception e) {
			log.error(e);
		}
	}

	private JmsTemplate createTopicJmsTemplate(String queueName) {
		JmsTemplate jmsTemplate;
		jmsTemplate = new JmsTemplate(cachingConnectionFactory);
		jmsTemplate.setDefaultDestinationName(queueName);
		jmsTemplate.setPubSubDomain(true);
		// jmsTemplate.setMessageConverter(messageConverter);
		jmsTemplate.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
		jmsTemplate.setSessionAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
		return jmsTemplate;
	}

	private JmsTemplate createQueueJmsTemplate(String queueName) {
		JmsTemplate jmsTemplate;
		jmsTemplate = new JmsTemplate(cachingConnectionFactory);
		jmsTemplate.setDefaultDestinationName(queueName);
		jmsTemplate.setPubSubDomain(false);
		// jmsTemplate.setMessageConverter(messageConverter);
		jmsTemplate.setDeliveryMode(DeliveryMode.PERSISTENT);
		jmsTemplate.setSessionAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
		jmsTemplate.setReceiveTimeout(10000);
		return jmsTemplate;
	}
}
